package t20240513;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.command.RedisCommand;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.streaming.connectors.redis.config.RedisValidator.REDIS_COMMAND;

public class RedisTest {

    public static void main(String[] args) {

        Configuration flinkConf = new Configuration();
        flinkConf.setString("rest.port","9091");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(flinkConf);
        env.setParallelism(2);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        tEnv.createTemporaryView("datagen", env.fromElements(
                Row.of("test_zadd", 100, "aaa", "aaa", "ddd"),
                Row.of("test_zadd", 100, "bbb", "bbb", "ddd"),
                Row.of("test_zadd", 100, "ccc", "ccc", "ddd"),
                Row.of("test_zadd", 100, "ddd", "aaa", "ccc")), "key, score, field, rem_begin, rem_end");


        String sinkSql =
                "  CREATE TABLE redis_sink (" +
                        "    key STRING," +
                        "    score INT," +
                        "    sn STRING," +
                        "    mins STRING," +
                        "    maxs STRING," +
                        "    primary key (key) not ENFORCED" +
                        "    ) WITH (" +
                        "    'connector' = 'redis'," +
                        "    'host' = '127.0.0.1'," +
                        "    'port' = '6379'," +
                        "    'redis-mode' = 'single'," +
                        "    'timeout' = '2000'," +
                        "    'password' = 'yy2024'," +
                        "    'command' = 'ZADD'," +
                        "    'zset.zremrangeby' = 'LEX'," +
                        "    'max.retries' = '5'" +
                        "  )";

        tEnv.executeSql(sinkSql);
        String sql = " insert into redis_sink select * from datagen";


        tEnv.executeSql(sql);

    }
}
